package com.yanfan.mqtt.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.yanfan.common.constant.YanfanConstant;
import com.yanfan.common.core.device.DeviceAndProtocol;
import com.yanfan.common.core.mq.DeviceReport;
import com.yanfan.common.core.mq.DeviceReportBo;
import com.yanfan.common.core.mq.MQSendMessageBo;
import com.yanfan.common.core.mq.message.*;
import com.yanfan.common.core.mq.ota.OtaUpgradeBo;
import com.yanfan.common.core.protocol.Message;
import com.yanfan.common.core.protocol.modbus.ModbusCode;
import com.yanfan.common.core.redis.RedisCache;
import com.yanfan.common.core.redis.RedisKeyBuilder;
import com.yanfan.common.core.thingsModel.ThingsModelSimpleItem;
import com.yanfan.common.enums.FunctionReplyStatus;
import com.yanfan.common.enums.OTAUpgrade;
import com.yanfan.common.enums.ServerType;
import com.yanfan.common.enums.TopicType;
import com.yanfan.common.exception.ServiceException;
import com.yanfan.common.utils.DateUtils;
import com.yanfan.common.utils.StringUtils;
import com.yanfan.common.utils.gateway.CRC16Utils;
import com.yanfan.common.utils.gateway.mq.TopicsUtils;
import com.yanfan.common.utils.ip.IpUtils;
import com.yanfan.common.utils.modbus.ModbusUtils;
import com.yanfan.iot.cache.IFirmwareCache;
import com.yanfan.iot.cache.ITSLCache;
import com.yanfan.iot.domain.*;
import com.yanfan.iot.enums.DeviceType;
import com.yanfan.iot.model.DeviceMqttConnectVO;
import com.yanfan.iot.model.NtpModel;
import com.yanfan.iot.model.ThingsModels.ThingsModelValueItem;
import com.yanfan.iot.model.ThingsModels.ValueItem;
import com.yanfan.iot.ruleEngine.MsgContext;
import com.yanfan.iot.ruleEngine.RuleProcess;
import com.yanfan.iot.service.*;
import com.yanfan.iot.util.SnowflakeIdWorker;
import com.yanfan.modbus.codec.ModbusMessageDecoder;
import com.yanfan.modbus.codec.ModbusMessageEncoder;
import com.yanfan.modbus.model.ModbusRtu;
import com.yanfan.mq.model.ReportDataBo;
import com.yanfan.mq.service.IDataHandler;
import com.yanfan.mq.service.IDeviceReportMessageService;
import com.yanfan.mq.service.IMqttMessagePublish;
import com.yanfan.mq.service.impl.MessageManager;
import com.yanfan.mqtt.manager.MqttRemoteManager;
import com.yanfan.mqtt.model.PushMessageBo;
import com.yanfan.mqttclient.PubMqttClient;
import com.yanfan.protocol.base.protocol.IProtocol;
import com.yanfan.protocol.domain.DeviceProtocol;
import com.yanfan.protocol.service.IProtocolManagerService;
import com.yanfan.sip.service.IGatewayService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

/**
 * 消息推送方法集合
 *
 * @author bill
 */
@Slf4j
@Service
public class MqttMessagePublishImpl implements IMqttMessagePublish {
    private static final ModbusMessageDecoder decoder = new ModbusMessageDecoder("com.yanfan.modbus");
    private static final ModbusMessageEncoder encoder = new ModbusMessageEncoder("com.yanfan.modbus");
    private final SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(3);
    @Resource
    private IProductService productService;
    @Resource
    private IProtocolManagerService protocolManagerService;
    @Resource
    private PubMqttClient mqttClient;
    @Resource
    private IFirmwareCache firmwareCache;
    @Resource
    private IFirmwareTaskDetailService firmwareTaskDetailService;
    @Resource
    private MessageManager messageManager;
    @Resource
    private TopicsUtils topicsUtils;
    @Resource
    private IDeviceService deviceService;
    @Resource
    private ISimulateLogService simulateLogService;
    @Resource
    private MqttRemoteManager remoteManager;
    @Resource
    private IDeviceReportMessageService reportMessageService;
    @Resource
    private IFunctionLogService functionLogService;
    @Resource
    private IDataHandler dataHandler;
    @Resource
    private IThingsModelService thingsModelService;
    @Resource
    private RuleProcess ruleProcess;
    @Autowired
    private IGatewayService gatewayService;
    @Resource
    private ITSLCache itslCache;
    @Resource
    private RedisCache redisCache;

    @Override
    public InstructionsMessage buildMessage(DeviceDownMessage downMessage, TopicType type) {
        /*返回的组将数据*/
        InstructionsMessage message = new InstructionsMessage();
        /*根据设备编号查询产品信息*/
        if (StringUtils.isEmpty(downMessage.getProtocolCode())) {
            Product product = productService.getProductBySerialNumber(downMessage.getSerialNumber());
            Optional.ofNullable(product).orElseThrow(() -> new ServiceException("产品为空"));
            downMessage.setProtocolCode(product.getProtocolCode());
        }
        String serialNumber = downMessage.getSerialNumber() == null ? "" : downMessage.getSerialNumber();

        /*组建Topic*/
        String topicName = "";
        if (downMessage.getServerType().equals(ServerType.MQTT)) {
            topicName = topicsUtils.buildTopic(downMessage.getProductId(), serialNumber, type);
        }
        /*获取编码协议*/
        IProtocol protocolInstance = protocolManagerService.getProtocolByProtocolCode(downMessage.getProtocolCode());
        DeviceData encodeData = DeviceData
                .builder()
                .downMessage(downMessage)
                .serialNumber(serialNumber)
                .body(downMessage.getBody())
                .code(downMessage.getCode())
                .topicName(topicName)
                .build();

        DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(serialNumber);
        Optional
                .ofNullable(deviceAndProtocol)
                .orElseThrow(() -> new ServiceException("服务下发的设备:[" + serialNumber + "]不存在"));

        MQSendMessageBo bo = new MQSendMessageBo();
        bo.setDp(deviceAndProtocol);
        bo.setShadow(false);
        bo.setSerialNumber(serialNumber);
        bo.setParams(new JSONObject());
        bo.setData(encodeData);
        //根据协议编码后数据
        FunctionCallBackBo data = protocolInstance.encode(bo);
        message.setMessage(data.getMessage());
        message.setSerialNumber(serialNumber);
        message.setTopicName(topicName);
        return message;
    }

    /**
     * 服务(指令)下发
     */
    @Override
    public void funcSend(MQSendMessageBo bo) {
        DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(bo.getSerialNumber());
        Optional
                .ofNullable(deviceAndProtocol)
                .orElseThrow(() -> new ServiceException("服务下发的设备:[" + bo.getSerialNumber() + "]不存在"));
        int deviceType = deviceAndProtocol.getDeviceType();
        String sourceNo = bo.getSerialNumber();
        //处理数据下发
        bo.setDp(deviceAndProtocol);
        Long productId = bo.getDp().getProductId();
        String serialNumber = bo.getSerialNumber();
        String transport = bo.getDp().getTransport();
        //点位是组态处理(注意点位的下划线不要成功_,负责会认为是一个数组的点位，需要处理的)
        ThingsModelValueItem thingModels = new ThingsModelValueItem();
        int i = bo.getIdentifier().indexOf("_");
        if (i >= 2) {
            //获取物模型
            if (i!=2) {
                int startIndex = bo.getIdentifier().indexOf("_") + 1; // 从第一个下划线后面开始
                int endIndex = bo.getIdentifier().length(); // 到最后一个下划线之前
                String field = bo.getIdentifier().substring(startIndex, endIndex); // 提取子字符串
                int startIndexs = field.indexOf("_") + 1;
                String field1 = field.substring(startIndexs, field.length());
                thingModels = itslCache.getSingleThingModels(productId, field1);
            } else {
                String field1 = bo.getIdentifier().replaceAll(".*_(.*)", "$1");
                thingModels = itslCache.getSingleThingModels(productId, field1);
            }
        }
        else {
            //获取物模型
             thingModels = itslCache.getSingleThingModels(productId, bo.getIdentifier());
        }
        ModbusConfig modbusConfig = thingModels.getConfig();

        if (!Objects.isNull(modbusConfig)) {
            thingModels.getConfig().setModbusCode(ModbusUtils.getModbusCode(modbusConfig.getType()));
            if (deviceType == DeviceType.SUB_GATEWAY.getCode()) {
                //这里获取绑定网关时，设置的子设备地址
                thingModels
                        .getConfig()
                        .setSlave(deviceAndProtocol.getSlaveId() == null ? deviceAndProtocol.getProSlaveId() : deviceAndProtocol.getSlaveId());
                //设置网关产品id，设备编号
                bo.setSerialNumber(deviceAndProtocol.getGwSerialNumber());
                bo.getDp().setProductId(deviceAndProtocol.getGwProductId());
                productId = deviceAndProtocol.getGwProductId();
                serialNumber = deviceAndProtocol.getGwSerialNumber();
            }
        }
        bo.setThingsModel(JSON.toJSONString(thingModels));
        Integer type = thingModels.getType();
        //处理影子模式
        this.hadndlerShadow(bo, type);
        //下发指令日志
        FunctionLog funcLog = this.handleLog(bo, thingModels.getName());
        ServerType serverType = ServerType.explain(transport);


        /*下发服务数据处理对象*/
        DeviceDownMessage downMessage = DeviceDownMessage
                .builder()
                .messageId(bo.getMessageId())
                .body(bo.getParams())
                .serialNumber(bo.getSerialNumber())
                .productId(productId)
                .timestamp(DateUtils.getTimestamp())
                .identifier(bo.getIdentifier())
                .code(deviceType == ModbusCode.Read01.getCode() ? ModbusCode.Write05 : ModbusCode.Write06)
                .serverType(serverType)
                .build();

        DeviceData encodeData = DeviceData
                .builder()
                .downMessage(downMessage)
                .serialNumber(serialNumber)
                .body(downMessage.getBody())
                .code(downMessage.getCode())
                .build();

        bo.setData(encodeData);

        //组建下发服务指令
        FunctionCallBackBo backBo = buildMessage(bo);
        switch (serverType) {
            case MQTT:
                //  规则引擎脚本处理,完成后返回结果
                MsgContext context = ruleProcess.processRuleScript(serialNumber, 2, backBo.getTopicName(), new String(backBo.getMessage()));
                if (!Objects.isNull(context) && StringUtils.isNotEmpty(context.getPayload()) && StringUtils.isNotEmpty(context.getTopic())) {
                    backBo.setTopicName(context.getTopic());
                    backBo.setMessage(context.getPayload().getBytes());
                }

                publishWithLog(backBo.getTopicName(), backBo.getMessage(), funcLog);
                log.debug("=>服务下发,topic=[{}],指令=[{}]", backBo.getTopicName(), new String(backBo.getMessage()));
                break;
            case TCP:
                Message data = new Message();
                data.setPayload(Unpooled.wrappedBuffer(backBo.getMessage()));
                data.setClientId(backBo.getSerialNumber());
                messageManager.requestR(serialNumber, data, Message.class);
                funcLog.setResultMsg(FunctionReplyStatus.NORELY.getMessage());
                funcLog.setResultCode(FunctionReplyStatus.NORELY.getCode());
                functionLogService.insertFunctionLog(funcLog);
                break;
            case UDP:
                break;
            case COAP:
                break;
            case GB28181:
                MqttMessagePublishImpl.log.debug("=>功能指令下发,functinos=[{}]", bo);
                gatewayService.sendFunction(serialNumber, bo.getIdentifier(), bo
                        .getParams()
                        .getString(bo.getIdentifier()));
                break;

        }
        //发送至前端数据调试
        String topic = topicsUtils.buildTopic(bo.getDp().getProductId(), bo.getSerialNumber(), TopicType.MESSAGE_POST);
        DeviceMessage deviceMessage = new DeviceMessage();
        deviceMessage.setMessage(backBo.getSources());
        deviceMessage.setTime(new Date());
        deviceMessage.setTopicName(TopicType.FUNCTION_GET.getTopicSuffix());
        byte[] bytes = JSONObject.toJSONString(deviceMessage).getBytes();
        publishWithLog(topic, bytes, null);

        if (deviceAndProtocol.getProtocolCode().equals(YanfanConstant.PROTOCOL.ModbusRtu)) {
            //这里做一个消息id标记消息下发顺序，如果设备指令带流水号，则不需要使用
            String cacheKey = RedisKeyBuilder.buildDownMessageIdCacheKey(sourceNo);
            redisCache.zSetAdd(cacheKey, backBo.getSources() + ":" + bo.getMessageId(), DateUtils.getTimestampSeconds());
        }
        //处理指令下发权限问题
        deviceService.updateByOrder(bo.getUserId(), deviceAndProtocol.getDeviceId());
    }


    /**
     * 处理影子模式
     */
    private void hadndlerShadow(MQSendMessageBo bo, int type) {
        //处理设备影子模式
        if (bo.isShadow()) {
            List<ThingsModelSimpleItem> dataList = new ArrayList<>();
            bo.getParams().forEach((key, value) -> {
                ThingsModelSimpleItem item = new ThingsModelSimpleItem();
                item.setId(key);
                item.setValue(value + "");
                dataList.add(item);
            });
            ReportDataBo dataBo = new ReportDataBo();
            dataBo
                    .setDataList(dataList)
                    .setProductId(bo.getDp().getProductId())
                    .setSerialNumber(bo.getSerialNumber())
                    .setRuleEngine(false)
                    .setShadow(true)
                    .setType(type);
            dataHandler.reportData(dataBo);
        }
    }

    /**
     * 处理下发指令日志
     *
     * @return
     */
    private FunctionLog handleLog(MQSendMessageBo message, String modelName) {
        /* 下发服务数据存储对象*/
        JSONObject params = message.getParams();
        String val = params.get(message.getIdentifier()) + "";
        message.setValue(val);
        FunctionLog funcLog = new FunctionLog();
        funcLog.setCreateTime(DateUtils.getNowDate());
        funcLog.setFunValue(val);
        funcLog.setMessageId(message.getMessageId());
        funcLog.setSerialNumber(message.getSerialNumber());
        funcLog.setIdentify(message.getIdentifier());
        funcLog.setShowValue(val);
        funcLog.setFunType(1);
        funcLog.setModelName(modelName);
        return funcLog;
    }

    /**
     * OTA升级下发
     *
     * @param bo
     */
    @Override
    public void upGradeOTA(OtaUpgradeBo bo) {
        //获取唯一messageId
        bo.setMessageId(String.valueOf(snowflakeIdWorker.nextId()));
        //获取固件版本缓存
        String url = this.firmwareCache.getFirmwareCache(bo.getOtaId());
        if (StringUtils.isEmpty(url)) {
            firmwareCache.setFirmwareCache(bo.getOtaId(), bo.getOtaUrl());
        }
        DeviceProtocol deviceProtocol = protocolManagerService.getProtocolBySerialNumber(bo.getSerialNumber());
        IProtocol protocol = deviceProtocol.getProtocol();
        /*组建下发OTA升级topic*/
        String topicName = topicsUtils.buildTopic(bo.getProductId(), bo.getSerialNumber(), TopicType.FIRMWARE_SET);
        DeviceDownMessage deviceDownMessage = buildMessage(bo);
        DeviceData deviceSource = DeviceData
                .builder()
                .serialNumber(bo.getSerialNumber())
                .topicName(topicName)
                .downMessage(deviceDownMessage)
                .build();
        //TODO -- 后续处理 。。编码OTA升级消息
        byte[] otaUpgrade = protocol.encodeOTA(deviceSource);
        FunctionLog log = new FunctionLog();
        log.setCreateTime(DateUtils.getNowDate());
        log.setSerialNumber(bo.getSerialNumber());
        log.setFunType(3);
        log.setMessageId(bo.getMessageId());
        log.setDeviceName(bo.getDeviceName());
        log.setIdentify("OTA");
        log.setFunValue("名称:" + bo.getFirmwareName() + " 版本:" + bo.getFirmwareVersion());
        log.setShowValue(bo.getOtaUrl());
        // 通过内部mqtt客户端下发消息
        publishWithLog(topicName, otaUpgrade, log);

        //    更新数据库
        firmwareTaskDetailService.update(bo, OTAUpgrade.SEND);

    }


    @Override
    public FunctionCallBackBo buildMessage(MQSendMessageBo bo) {
        String protocolCode = bo.getDp().getProtocolCode();
        Long productId = bo.getDp().getProductId();
        String serialNumber = bo.getSerialNumber();
        /*组建Topic*/
        String topic = topicsUtils.buildTopic(productId, serialNumber, TopicType.FUNCTION_GET);
        bo.setTopicName(topic);
        /*获取编码协议*/
        IProtocol protocolInstance = protocolManagerService.getProtocolByProtocolCode(protocolCode);
        //根据协议编码后数据
        FunctionCallBackBo callBackBo = protocolInstance.encode(bo);
        callBackBo.setSerialNumber(serialNumber);
        callBackBo.setTopicName(topic);
        return callBackBo;
    }

    @Override
    public void sendFunctionMessage(DeviceReportBo bo) {
        log.warn("=>功能指令下发,sendFunctionMessage bo=[{}]", bo);
        Device device = deviceService.selectDeviceBySerialNumber(bo.getSerialNumber());
        Optional
                .ofNullable(device)
                .orElseThrow(() -> new ServiceException("服务下发的设备:[" + bo.getSerialNumber() + "]不存在"));

        Product product = productService.selectProductByProductId(topicsUtils.parseProductId(bo.getTopicName()));
        ServerType serverType = ServerType.explain(product.getTransport());
        Optional.ofNullable(serverType).orElseThrow(() -> new ServiceException("产品的传输协议编码为空!"));

        switch (serverType) {
            case GB28181:
                List<ThingsModelSimpleItem> functinos = JSON.parseArray(bo.getData(), ThingsModelSimpleItem.class);
                log.debug("=>功能指令下发,functinos=[{}]", functinos.toString());
                gatewayService.sendFunction(bo.getSerialNumber(), functinos);
                break;
        }
    }

    /**
     * 1.发布设备状态
     */
    @Override
    public void publishStatus(Long productId, String deviceNum, int deviceStatus, int isShadow, int rssi) {
        String message = "{\"status\":" + deviceStatus + ",\"isShadow\":" + isShadow + ",\"rssi\":" + rssi + "}";
        String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.STATUS_POST);
        mqttClient.publish(1, false, topic, message);
    }


    /**
     * 2.发布设备信息
     */
    @Override
    public void publishInfo(Long productId, String deviceNum) {
        String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.INFO_GET);
        mqttClient.publish(1, false, topic, "");
    }

    /**
     * 3.发布时钟同步信息
     *
     * @param bo 数据模型
     */
    public void publishNtp(ReportDataBo bo) {
        NtpModel ntpModel = JSON.parseObject(bo.getMessage(), NtpModel.class);
        ntpModel.setServerRecvTime(System.currentTimeMillis());
        ntpModel.setServerSendTime(System.currentTimeMillis());
        String topic = topicsUtils.buildTopic(bo.getProductId(), bo.getSerialNumber(), TopicType.NTP_GET);
        mqttClient.publish(1, false, topic, JSON.toJSONString(ntpModel));
    }

    /**
     * 4.发布属性
     * delay 延时，秒为单位
     */
    @Override
    public void publishProperty(Long productId, String deviceNum, List<ThingsModelSimpleItem> thingsList, int delay) {
        String pre = "";
        if (delay > 0) {
            pre = "$delayed/" + delay + "/";
        }
        String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.FUNCTION_GET);
        if (thingsList == null) {
            mqttClient.publish(1, true, topic, "");
        } else {
            mqttClient.publish(1, true, topic, JSON.toJSONString(thingsList));
        }
    }

    /**
     * 5.发布功能
     * delay 延时，秒为单位
     */
    @Override
    public void publishFunction(Long productId, String deviceNum, List<ThingsModelSimpleItem> thingsList, int delay) {
        String pre = "";
        if (delay > 0) {
            pre = "$delayed/" + delay + "/";
        }
        String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.FUNCTION_GET);
        if (thingsList == null) {
            mqttClient.publish(1, true, topic, "");
        } else {
            mqttClient.publish(1, true, topic, JSON.toJSONString(thingsList));
        }

    }

    /**
     * 设备数据同步
     *
     * @param deviceNumber 设备编号
     * @return 设备
     */
    public Device deviceSynchronization(String deviceNumber) {
        Device device = deviceService.selectDeviceBySerialNumber(deviceNumber);
        // 1-未激活，2-禁用，3-在线，4-离线
        if (device.getStatus() == 3) {
            device.setStatus(4);
            deviceService.updateDeviceStatus(device);
            // 发布设备信息
            publishInfo(device.getProductId(), device.getSerialNumber());
        }
        return device;
    }

    /**
     * 用于发送模拟客户端数据
     */
    @Override
    public void sendSimulatorMessage(String topic, byte[] source) {
        String serialNumber = topicsUtils.parseSerialNumber(topic);
        if (serialNumber.contains("_")) {
            serialNumber = serialNumber.split("_")[0];
        }
        Long productId = topicsUtils.parseProductId(topic);
        DeviceData data = DeviceData
                .builder()
                .serialNumber(serialNumber)
                .topicName(topic)
                .data(source)
                .isEnabledTest(true)
                .buf(Unpooled.wrappedBuffer(source))
                .build();
        //获取下发的读指令
        ModbusRtu rtu = decoder.decode(data);
        /*根据数据个数，拼装随机数,模拟modbus设备上报的数据*/
        short[] bytes = new short[rtu.getCount()];
        Random random = new Random();

        DeviceReport report = new DeviceReport();
        List<ThingsModelSimpleItem> values = new ArrayList<>();
        for (int i = 0; i < rtu.getCount(); i++) {
            int nextInt = random.nextInt(100);
            bytes[i] = (short) nextInt;
            ThingsModelSimpleItem things = new ThingsModelSimpleItem();
            things.setId(rtu.getAddress() + i + "");
            things.setValue(nextInt + "");
            things.setTs(DateUtils.getNowDate());
            things.setSlaveId(rtu.getSlaveId());
            values.add(things);
        }
        report.setSlaveId(rtu.getSlaveId());
        report.setThingsModelSimpleItem(values);
        report.setSerialNumber(serialNumber);
        report.setProductId(productId);
        report.setServerType(ServerType.MQTT);
        String upTopic = topicsUtils.buildTopic(productId, serialNumber, TopicType.PROPERTY_POST_SIMULATE);
        reportMessageService.processNoSub(report, upTopic);

        rtu.setData(bytes);
//        rtu.setEnableTest(true);
        ByteBuf buf = encoder.encode(rtu);
        byte[] write = new byte[buf.writerIndex()];
        buf.readBytes(write);
        byte[] result = CRC(write);
        ReferenceCountUtil.release(buf);
        log.debug("<<<<<<<<<<模拟客户端推送主题[{}],消息[{}]", upTopic, ByteBufUtil.hexDump(result));


        MqttBo send = new MqttBo();
        send.setData(ByteBufUtil.hexDump(source));
        send.setTopic(topic);
        MqttBo receive = new MqttBo();
        receive.setTopic(upTopic);
        receive.setData(ByteBufUtil.hexDump(result));
        updateSimulationLog(send, receive, serialNumber);
        //发送至WS
        String pushTopic = topicsUtils.buildTopic(productId, serialNumber, TopicType.WS_SERVICE_INVOKE_SIMULATE);
        sendSimulationWs(send, receive, pushTopic);
    }

    /**
     * 模拟设备写客户端数据
     *
     * @param topic
     */
    @Override
    public void getSimulatorInfo(String topic, byte[] source) {
        String serialNumber = topicsUtils.parseSerialNumber(topic);
        Long productId = topicsUtils.parseProductId(topic);
        DeviceData data = DeviceData
                .builder()
                .serialNumber(serialNumber)
                .topicName(topic)
                .data(source)
                .code(ModbusCode.Write06)
                .buf(Unpooled.wrappedBuffer(source))
                .build();
        //获取下发的写指令
        ModbusRtu rtu = decoder.decode(data);
        //更新redis的值
        String cacheKey = RedisKeyBuilder.buildTSLVCacheKey(productId, serialNumber);
        String hkey = rtu.getAddress() + "#" + rtu.getSlaveId();
        ValueItem modbusItem = redisCache.getCacheMapValue(cacheKey, hkey);
        String s1 = rtu.getAddress() + "";
        if (null == modbusItem) {
            modbusItem = new ValueItem();
            modbusItem.setId(s1);
            modbusItem.setSlaveId(rtu.getSlaveId());
            modbusItem.setRegArr(s1);
        }
        modbusItem.setShadow(rtu.getWriteData() + "");
        modbusItem.setValue(rtu.getWriteData() + "");
        modbusItem.setTs(DateUtils.getNowDate());
        redisCache.setCacheMapValue(cacheKey, hkey, com.alibaba.fastjson2.JSONObject.toJSONString(modbusItem));

        /*将最新值推送到前端*/
        List<ThingsModelSimpleItem> result = new ArrayList<>();
        ThingsModelSimpleItem item = new ThingsModelSimpleItem();
        item.setId(s1);
        item.setValue(rtu.getWriteData() + "");
        item.setTs(DateUtils.getNowDate());
        item.setSlaveId(rtu.getSlaveId());
        result.add(item);
        PushMessageBo push = new PushMessageBo();
        push.setTopic(topicsUtils.buildTopic(productId, serialNumber, TopicType.WS_SERVICE_INVOKE));
        push.setMessage(JSON.toJSONString(result));
        remoteManager.pushCommon(push);

        //指令原路返回
        String[] s = serialNumber.split("_");
        if (s.length > 1) {
            serialNumber = s[0];
        }
        PushMessageBo messageBo = new PushMessageBo();
        JSONArray array = new JSONArray();
        String upTopic = topicsUtils.buildTopic(productId, serialNumber, TopicType.WS_SERVICE_INVOKE_SIMULATE);
        MqttBo receive = new MqttBo();
        receive.setTopic(upTopic);
        receive.setData(ByteBufUtil.hexDump(source));
        receive.setDirection("receive");
        array.add(receive);
        messageBo.setMessage(array.toJSONString());
        messageBo.setTopic(upTopic);
        remoteManager.pushCommon(messageBo);

    }


    /**
     * 模拟设备数据更新
     */
    public void updateSimulationLog(MqttBo send, MqttBo receive, String serialNumber) {
        SimulateLog log = new SimulateLog();
        log.setSendData(JSON.toJSONString(send));
        log.setCallbackData(JSON.toJSONString(receive));
        log.setSerialNumber(serialNumber);
        log.setCreateTime(DateUtils.getNowDate());
        simulateLogService.insertSimulateLog(log);
    }

    /**
     * 发送模拟设备到WS
     */
    public void sendSimulationWs(MqttBo send, MqttBo receive, String topic) {
        PushMessageBo messageBo = new PushMessageBo();
        messageBo.setTopic(topic);
        JSONArray array = new JSONArray();
        send.setDirection("send");
        send.setTs(DateUtils.getNowDate());
        receive.setTs(DateUtils.getNowDate());
        receive.setDirection("receive");
        array.add(send);
        array.add(receive);
        messageBo.setMessage(array.toJSONString());
        remoteManager.pushCommon(messageBo);
    }

    public byte[] CRC(byte[] source) {
        source[2] = (byte) ((int) source[2] * 2);
        byte[] result = new byte[source.length + 2];
        byte[] crc16Byte = CRC16Utils.getCrc16Byte(source);
        System.arraycopy(source, 0, result, 0, source.length);
        System.arraycopy(crc16Byte, 0, result, result.length - 2, 2);
        return result;
    }


    /**
     * =======
     * >>>>>>> guanfeng/dev
     * 搭建消息
     *
     * @param bo
     * @return
     */
    private DeviceDownMessage buildMessage(OtaUpgradeBo bo) {
        String messageId = String.valueOf(snowflakeIdWorker.nextId());
        bo.setMessageId(messageId);
        bo.setOtaUrl("http://" + IpUtils.getHostIp() + bo.getOtaUrl());
        return DeviceDownMessage
                .builder()
                .productId(bo.getProductId())
                .serialNumber(bo.getSerialNumber())
                .body(JSON.toJSON(bo))
                .timestamp(DateUtils.getTimestamp())
                .messageId(messageId)
                .build();

    }

    public void publishWithLog(String topic, byte[] pushMessage, FunctionLog log) {
        try {
            redisCache.incr2(YanfanConstant.REDIS.MESSAGE_SEND_TOTAL, -1L);
            redisCache.incr2(YanfanConstant.REDIS.MESSAGE_SEND_TODAY, 60 * 60 * 24);
            String serialNumber = topicsUtils.parseSerialNumber(topic);
            DeviceMqttConnectVO deviceMqttConnect = deviceService.getDeviceMqttConnect(null, serialNumber,null);
            if (deviceMqttConnect == null) {
                System.out.println("设备不存在："+serialNumber);
            }
            else{
                if(deviceMqttConnect.getMqttstats()==1){
                    mqttClient.publish(pushMessage, deviceMqttConnect.getDyname(), false, 0);
                }
                else {
                    mqttClient.publish(pushMessage, topic, false, 0);
                }
            }
            if (null != log) {
                //存储服务下发成功
                log.setResultMsg(FunctionReplyStatus.NORELY.getMessage());
                log.setResultCode(FunctionReplyStatus.NORELY.getCode());
                functionLogService.insertFunctionLog(log);
            }
        } catch (Exception e) {
            if (null != log) {
                //服务下发失败存储
                log.setResultMsg(FunctionReplyStatus.FAIl.getMessage() + "原因: " + e.getMessage());
                log.setResultCode(FunctionReplyStatus.FAIl.getCode());
                functionLogService.insertFunctionLog(log);
            }
        }
    }

}
